package minglue.util
/**
 * kafka的偏移量用redis来管理
 */
import org.slf4j.LoggerFactory
import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory
import java.io.File
import org.apache.kafka.common.TopicPartition
import java.util
object RedisOffSet { //函数作为成员函数
  def apply(groupId: String) = {
    //创建MAP形式的topic partition Offset
    var offset = Map[TopicPartition, Long]()
    //查询出redis中的所有topic partition
    val topicPartitionOffset: util.Map[String, String] = RedisCluster.hgetAll(groupId)
    // 导入java 隐士转换
    import scala.collection.JavaConversions._
    //将Redis中Topic下的partition中的offset 转为list
    val topicPartitionOffsetList: List[(String, String)] = topicPartitionOffset.toList
    //循环处理所有数据

    for (topicPl <- topicPartitionOffsetList) {
      val split: Array[String] = topicPl._1.split("[-]")
    		  offset += ( new TopicPartition( split(0), split(1).toInt) ->topicPl._2.toLong)
    }
    offset
  }
}
   